package cn.dmp.tools

import cn.dmp.utils.{NumberFormat, SchemaUtils}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SQLContext}
import org.apache.spark.sql.types._
import org.apache.spark.{SparkConf, SparkContext}

/**
  *
  * 将原始文件日志转换成parquet文件格式
  * 采用snappy压缩格式
  *
  */
// 用到了自己写的 cn.dmp.utils.{NumberFormat, SchemaUtils}这两个类
object Bzip2Parquet {
  def main(args: Array[String]): Unit = {
    // 0.检验参数的个数。最低需要两个：日志输入目录，日志输出目录。
    if (args.length != 2) {
      println(
        """
          |cn.dmp.tools.Bzip2Parquet
          |参数：
          | logInputPath
          | resultOutputPath
        """.stripMargin)
      sys.exit()
    }

    // 1.接收程序的参数。
    val Array(logInputPath, resultOutputPath) = args

    // 2.创建sparkconf->sparkContext。//s"${this.getClass.getSimpleName}"得到大明湖起哦按类的名称。
    val conf = new SparkConf().setAppName(s"${this.getClass.getSimpleName}").setMaster("local[*]")
    // RDD序列化到磁盘，worker与worker之间的传输。
    /**
      * spark优化：spark.serializer修改序列化方式
      * 进行节点的数据传递，或者保存数据时都会进行序列化。
      * spark默认的是org.apache.spark.serializer.JavaSerializer。
      * 而我们要修改成org.apache.spark.serializer.KryoSerializer。
      */
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    //输出数据的时候的压缩编码设置为snappy,spark2.0默认就是snappy。可以为<snappy,gzip,lzo>
    conf.set("spark.sql.parquet.compression.codec", "snappy")

    val sc = new SparkContext(conf)

    val sQLContext = new SQLContext(sc)


    // 3.读取日志数据
    val rawdata: RDD[String] = sc.textFile(logInputPath)

    // 4.根据业务需求对数据进行ETL:ETL是将业务系统的数据经过抽取、清洗转换之后加载到数据仓库的过程
    //xxx,x,x,x,,,,比如有这样的8个数据，用split(",")来切的话，后面这几个,,,的获取不到。必须用line.split(",",line.length或者-1)代表一直切分到这一行的最后。
    val dataRow: RDD[Row] = rawdata
      .map(line => line.split(",", line.length))
      .filter(_.length >= 85)
      .map(arr => {
        //import org.apache.spark.sql.Row 这个包还不好导，先写Row，然后鼠标移走，在移过来，alt+enter。然后选import class找到这个spark.sql.Row的包
        //将数组arr转换成spark的Row
        Row(
          arr(0),
          NumberFormat.toInt(arr(1)),
          NumberFormat.toInt(arr(2)),
          NumberFormat.toInt(arr(3)),
          NumberFormat.toInt(arr(4)),
          arr(5),
          arr(6),
          NumberFormat.toInt(arr(7)),
          NumberFormat.toInt(arr(8)),
          NumberFormat.toDouble(arr(9)),
          NumberFormat.toDouble(arr(10)),
          arr(11),
          arr(12),
          arr(13),
          arr(14),
          arr(15),
          arr(16),
          NumberFormat.toInt(arr(17)),
          arr(18),
          arr(19),
          NumberFormat.toInt(arr(20)),
          NumberFormat.toInt(arr(21)),
          arr(22),
          arr(23),
          arr(24),
          arr(25),
          NumberFormat.toInt(arr(26)),
          arr(27),
          NumberFormat.toInt(arr(28)),
          arr(29),
          NumberFormat.toInt(arr(30)),
          NumberFormat.toInt(arr(31)),
          NumberFormat.toInt(arr(32)),
          arr(33),
          NumberFormat.toInt(arr(34)),
          NumberFormat.toInt(arr(35)),
          NumberFormat.toInt(arr(36)),
          arr(37),
          NumberFormat.toInt(arr(38)),
          NumberFormat.toInt(arr(39)),
          NumberFormat.toDouble(arr(40)),
          NumberFormat.toDouble(arr(41)),
          NumberFormat.toInt(arr(42)),
          arr(43),
          NumberFormat.toDouble(arr(44)),
          NumberFormat.toDouble(arr(45)),
          arr(46),
          arr(47),
          arr(48),
          arr(49),
          arr(50),
          arr(51),
          arr(52),
          arr(53),
          arr(54),
          arr(55),
          arr(56),
          NumberFormat.toInt(arr(57)),
          NumberFormat.toDouble(arr(58)),
          NumberFormat.toInt(arr(59)),
          NumberFormat.toInt(arr(60)),
          arr(61),
          arr(62),
          arr(63),
          arr(64),
          arr(65),
          arr(66),
          arr(67),
          arr(68),
          arr(69),
          arr(70),
          arr(71),
          arr(72),
          NumberFormat.toInt(arr(73)),
          NumberFormat.toDouble(arr(74)),
          NumberFormat.toDouble(arr(75)),
          NumberFormat.toDouble(arr(76)),
          NumberFormat.toDouble(arr(77)),
          NumberFormat.toDouble(arr(78)),
          arr(79),
          arr(80),
          arr(81),
          arr(82),
          arr(83),
          NumberFormat.toInt(arr(84))
        )
      })

    // 5.将结果存储到本地磁盘!  注意：sQLContext.createDataFrame方法的两个参数：RDD[Row]数据 和 StructType类型要匹配。
    val dataFrame: DataFrame = sQLContext.createDataFrame(dataRow, SchemaUtils.logStructType)
    dataFrame.write.parquet(resultOutputPath)


    // 6.关闭sc
    sc.stop()

    //本地测试运行：!!!!!注意提前设置好参数：输入目录和输出目录
    /**
      * 输入目录：E:\大数据培训视频\项目+spark面试+flink+项目\项目\DMP\data\2016-10-01_06_p1_invalid.1475274123982.log.FINISH
      * 输出目录: E:\大数据培训视频\项目+spark面试+flink+项目\项目\DMP\biz2parquet
      *
      */

  }
}
